#ifndef __M_CHANNEL_H__
#define __M_CHANNEL_H__
#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "../mqcommon/mq_logger.hpp"
#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include "../mqcommon/mq_proto.pb.h"
#include "../mqcommon/mq_threadpool.hpp"
#include "mq_consumer.hpp"
#include "mq_host.hpp"
#include "mq_route.hpp"

namespace zdfmq
{
    // 类型别名定义：简化常用智能指针的使用
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using openChannelRequestPtr = std::shared_ptr<openChannelRequest>;
    using closeChannelRequestPtr = std::shared_ptr<closeChannelRequest>;
    using declareExchangeRequestPtr = std::shared_ptr<declareExchangeRequest>;
    using deleteExchangeRequestPtr = std::shared_ptr<deleteExchangeRequest>;
    using declareQueueRequestPtr = std::shared_ptr<declareQueueRequest>;
    using deleteQueueRequestPtr = std::shared_ptr<deleteQueueRequest>;
    using queueBindRequestPtr = std::shared_ptr<queueBindRequest>;
    using queueUnBindRequestPtr = std::shared_ptr<queueUnBindRequest>;
    using basicPublishRequestPtr = std::shared_ptr<basicPublishRequest>;
    using basicAckRequestPtr = std::shared_ptr<basicAckRequest>;
    using basicConsumeRequestPtr = std::shared_ptr<basicConsumeRequest>;
    using basicCancelRequestPtr = std::shared_ptr<basicCancelRequest>;

    /**
     * @brief 信道类，作为TCP连接与消息队列操作的中间层
     * 负责处理交换机、队列、消息发布/订阅等核心操作
     */
    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>; // 自身智能指针类型

        /**
         * @brief 构造函数
         * @param id 信道ID
         * @param host 所属虚拟主机
         * @param cmp 消费者管理器
         * @param codec Protobuf编解码器
         * @param conn TCP连接
         * @param pool 线程池（用于异步任务）
         */
        Channel(const std::string &id,
                const VirtualHost::ptr &host,
                const ConsumerManager::ptr &cmp,
                const ProtobufCodecPtr &codec,
                const muduo::net::TcpConnectionPtr &conn,
                const threadpool::ptr &pool) : _cid(id),
                                               _conn(conn),
                                               _codec(codec),
                                               _cmp(cmp),
                                               _host(host),
                                               _pool(pool)
        {
            DLOG("new Channel: %p", this);
        }

        /**
         * @brief 析构函数
         * 清理资源，移除关联的消费者
         */
        ~Channel()
        {
            if (_consumer.get() != nullptr)
            {
                _cmp->remove(_consumer->tag, _consumer->qname);
            }
            DLOG("del Channel: %p", this);
        }

        // 交换机操作
        /**
         * @brief 声明交换机
         * @param req 声明交换机请求
         */
        void declareExchange(const declareExchangeRequestPtr &req)
        {
            bool ret = _host->declareExchange(req->exchange_name(),
                                              req->exchange_type(), req->durable(),
                                              req->auto_delete(), req->args());
            return basicResponse(ret, req->rid(), req->cid());
        }

        /**
         * @brief 删除交换机
         * @param req 删除交换机请求
         */
        void deleteExchange(const deleteExchangeRequestPtr &req)
        {
            _host->deleteExchange(req->exchange_name());
            return basicResponse(true, req->rid(), req->cid());
        }

        // 队列操作
        /**
         * @brief 声明队列
         * @param req 声明队列请求
         */
        void declareQueue(const declareQueueRequestPtr &req)
        {
            bool ret = _host->declareQueue(req->queue_name(),
                                           req->durable(), req->exclusive(),
                                           req->auto_delete(), req->args());
            if (ret == false)
            {
                return basicResponse(false, req->rid(), req->cid());
            }
            _cmp->initQueueConsumer(req->queue_name()); // 初始化队列的消费者管理
            return basicResponse(true, req->rid(), req->cid());
        }

        /**
         * @brief 删除队列
         * @param req 删除队列请求
         */
        void deleteQueue(const deleteQueueRequestPtr &req)
        {
            _cmp->destroyQueueConsumer(req->queue_name()); // 先销毁队列的消费者管理
            _host->deleteQueue(req->queue_name());
            return basicResponse(true, req->rid(), req->cid());
        }

        // 队列与交换机绑定操作
        /**
         * @brief 绑定队列到交换机
         * @param req 绑定请求
         */
        void queueBind(const queueBindRequestPtr &req)
        {
            bool ret = _host->bind(req->exchange_name(),
                                   req->queue_name(), req->binding_key());
            return basicResponse(ret, req->rid(), req->cid());
        }

        /**
         * @brief 解除队列与交换机的绑定
         * @param req 解绑请求
         */
        void queueUnBind(const queueUnBindRequestPtr &req)
        {
            _host->unBind(req->exchange_name(), req->queue_name());
            return basicResponse(true, req->rid(), req->cid());
        }

        // 消息发布
        /**
         * @brief 发布消息
         * @param req 发布消息请求
         * 流程：检查交换机 -> 路由匹配队列 -> 发送消息到队列 -> 异步推送消息给消费者
         */
        void basicPublish(const basicPublishRequestPtr &req)
        {
            // 1. 判断交换机是否存在
            auto ep = _host->selectExchange(req->exchange_name());
            if (ep.get() == nullptr)
            {
                return basicResponse(false, req->rid(), req->cid());
            }

            // 2. 路由匹配目标队列
            MsgQueueBindingMap mqbm = _host->exchangeBindings(req->exchange_name());
            BasicProperties *properties = nullptr;
            std::string routing_key;
            if (req->has_properties())
            {
                properties = req->mutable_properties();
                routing_key = properties->routing_key();
            }

            for (auto &binding : mqbm)
            {
                if (Router::route(ep->type, routing_key, binding.second->binding_key))
                {
                    // 3. 将消息添加到队列
                    _host->basicPublish(binding.first, properties, req->body());
                    // 4. 异步推送消息给消费者
                    auto task = std::bind(&Channel::consume, this, binding.first);
                    _pool->push(task);
                }
            }

            return basicResponse(true, req->rid(), req->cid());
        }

        // 消息确认
        /**
         * @brief 确认消息已消费（Ack）
         * @param req 确认请求
         */
        void basicAck(const basicAckRequestPtr &req)
        {
            _host->basicAck(req->queue_name(), req->message_id());
            return basicResponse(true, req->rid(), req->cid());
        }

        // 消息订阅与取消
        /**
         * @brief 订阅队列消息
         * @param req 订阅请求
         * 流程：检查队列存在性 -> 创建消费者 -> 注册回调函数
         */
        void basicConsume(const basicConsumeRequestPtr &req)
        {
            // 1. 判断队列是否存在
            bool ret = _host->existsQueue(req->queue_name());
            if (ret == false)
            {
                return basicResponse(false, req->rid(), req->cid());
            }

            // 2. 创建消息推送回调
            auto cb = std::bind(&Channel::callback, this, std::placeholders::_1,
                                std::placeholders::_2, std::placeholders::_3);

            // 3. 创建消费者（当前信道作为消费者载体）
            _consumer = _cmp->create(req->consumer_tag(), req->queue_name(), req->auto_ack(), cb);
            return basicResponse(true, req->rid(), req->cid());
        }

        /**
         * @brief 取消订阅
         * @param req 取消订阅请求
         */
        void basicCancel(const basicCancelRequestPtr &req)
        {
            _cmp->remove(req->consumer_tag(), req->queue_name());
            return basicResponse(true, req->rid(), req->cid());
        }

    private:
        /**
         * @brief 消息推送回调函数
         * 当消费者收到消息时，通过TCP连接推送给客户端
         * @param tag 消费者标签
         * @param bp 消息属性
         * @param body 消息体
         */
        void callback(const std::string tag, const BasicProperties *bp, const std::string &body)
        {
            basicConsumeResponse resp;
            resp.set_cid(_cid);
            resp.set_body(body);
            resp.set_consumer_tag(tag);
            if (bp)
            {
                resp.mutable_properties()->set_id(bp->id());
                resp.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                resp.mutable_properties()->set_routing_key(bp->routing_key());
            }
            _codec->send(_conn, resp); // 发送消息给客户端
        }

        /**
         * @brief 消费消息任务（由线程池调用）
         * 从队列取消息并推送给订阅者
         * @param qname 队列名
         */
        void consume(const std::string &qname)
        {
            // 1. 从队列取消息
            MessagePtr mp = _host->basicConsume(qname);
            if (mp.get() == nullptr)
            {
                DLOG("执行消费任务失败，%s 队列没有消息！", qname.c_str());
                return;
            }

            // 2. 选择订阅该队列的消费者
            Consumer::ptr cp = _cmp->choose(qname);
            if (cp.get() == nullptr)
            {
                DLOG("执行消费任务失败，%s 队列没有消费者！", qname.c_str());
                return;
            }

            // 3. 推送消息给消费者（触发callback回调）
            cp->callback(cp->tag, mp->mutable_payload()->mutable_properties(), mp->payload().body());

            // 4. 自动确认模式下直接删除消息
            if (cp->auto_ack)
                _host->basicAck(qname, mp->payload().properties().id());
        }

        /**
         * @brief 发送基础响应给客户端
         * @param ok 操作结果（成功/失败）
         * @param rid 请求ID（用于匹配请求与响应）
         * @param cid 信道ID
         */
        void basicResponse(bool ok, const std::string &rid, const std::string &cid)
        {
            basicCommonResponse resp;
            resp.set_rid(rid);
            resp.set_cid(cid);
            resp.set_ok(ok);
            _codec->send(_conn, resp);
        }

    private:
        std::string _cid;                   // 信道ID
        Consumer::ptr _consumer;            // 关联的消费者（若为订阅者）
        muduo::net::TcpConnectionPtr _conn; // 对应的TCP连接
        ProtobufCodecPtr _codec;            // Protobuf编解码器
        ConsumerManager::ptr _cmp;          // 消费者管理器
        VirtualHost::ptr _host;             // 所属虚拟主机
        threadpool::ptr _pool;              // 线程池（用于异步任务）
    };

    /**
     * @brief 信道管理器
     * 负责管理所有信道的创建、销毁和查找
     */
    class ChannelManager
    {
    public:
        using ptr = std::shared_ptr<ChannelManager>; // 自身智能指针类型

        ChannelManager() {} // 构造函数

        /**
         * @brief 创建新信道
         * @param id 信道ID
         * @param host 所属虚拟主机
         * @param cmp 消费者管理器
         * @param codec 编解码器
         * @param conn TCP连接
         * @param pool 线程池
         * @return 是否创建成功（ID已存在则失败）
         */
        bool openChannel(const std::string &id,
                         const VirtualHost::ptr &host,
                         const ConsumerManager::ptr &cmp,
                         const ProtobufCodecPtr &codec,
                         const muduo::net::TcpConnectionPtr &conn,
                         const threadpool::ptr &pool)
        {
            std::unique_lock<std::mutex> lock(_mutex); // 线程安全锁
            auto it = _channels.find(id);
            if (it != _channels.end())
            { // 信道ID已存在
                DLOG("信道：%s 已经存在!", id.c_str());
                return false;
            }
            // 创建新信道并加入管理
            auto channel = std::make_shared<Channel>(id, host, cmp, codec, conn, pool);
            _channels.insert(std::make_pair(id, channel));
            return true;
        }

        /**
         * @brief 关闭信道（从管理器中移除）
         * @param id 信道ID
         */
        void closeChannel(const std::string &id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _channels.erase(id);
        }

        /**
         * @brief 获取信道
         * @param id 信道ID
         * @return 信道智能指针（不存在则返回空）
         */
        Channel::ptr getChannel(const std::string &id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(id);
            if (it == _channels.end())
            {
                return Channel::ptr();
            }
            return it->second;
        }

    private:
        std::mutex _mutex; // 线程安全锁
        // 信道映射表：key=信道ID，value=信道智能指针
        std::unordered_map<std::string, Channel::ptr> _channels;
    };
}

#endif